package com.zhj.kafka.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.zhj.kafka.KafkaProducerApplication;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    public void send(Object msg) throws JsonProcessingException {
        String jsonString = KafkaProducerApplication.objectMapper.writeValueAsString(msg);
        log.info("准备发送消息："+jsonString);
        ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send("test-1", jsonString);
        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.info("消息发送失败:"+throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                log.info("消息发送成功："+stringObjectSendResult.toString());
            }
        });
    }



}
